package cn.ivicar.redis.pubsub;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.codehaus.plexus.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.ivicar.redis.JedisClusterFactory;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPubSub;

public class RedisConsumer extends Thread {

	private static transient final Logger LOGGER = LoggerFactory.getLogger(RedisSubThread.class);

	private JedisPubSub subscriber;

	private String channel;

	private MessageHander hander;
	
	private RedisSubThread redisSubThread = null;

	private static Map<String, Stat> consumerStat = new HashMap<String, Stat>();
	
	private static ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);

	static {
		// 5分钟执行一次
		ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
		service.scheduleAtFixedRate(new Runnable() {
			public void run() {
				Set<String> keys = consumerStat.keySet();
				for (String key : keys) {
					Stat stat = consumerStat.get(key);
					String msg = String.format(
							"redis channel [%s] consume message in last 5 minutes: success:[%s] fail:[%s] total:[%s]",
							key, stat.getAndResetSuccess(), stat.getAndResetFail(), stat.getTotal());
					LOGGER.warn(msg);
				}
			}
		}, 0, 300, TimeUnit.SECONDS);
	}

	public RedisConsumer(String channel, MessageHander hander) {
		this.channel = channel;
		this.hander = hander;
		
	}

	public boolean isOpened() {
		return subscriber != null;
	}

	public void open() {
		//subscriber = new RedisSubscriber();
		//redisSubThread = new RedisSubThread(this.channel, this.subscriber);
		//redisSubThread.start();
		consumerStat.put(this.channel, new Stat());
		
		timer.scheduleAtFixedRate(new Runnable() {
			public void run() {
				boolean have = true;
				while(have){
					try {
						
						JedisCluster cluster = JedisClusterFactory.getDefaultJedisCluster();
						String message = cluster.lpop(RedisConsumer.this.channel);
						have = StringUtils.isNotBlank(message) ? true : false;
						if(have){
							LOGGER.info(String.format("receive redis published message, channel %s, message %s", channel, message));
							RedisConsumer.this.hander.handle(message);
							consumerStat.get(channel).increaseSuccess();
						}
						
					} catch (Exception e) {
						e.printStackTrace();
						consumerStat.get(channel).increaseFail();
					} finally {
						consumerStat.get(channel).increaseTotal();
					}
				}
				
			}
		}, 0, 15, TimeUnit.SECONDS);

	}

	class RedisSubThread extends Thread {

		private JedisPubSub subscriber;

		private String channel;

		public RedisSubThread(String channel, JedisPubSub subscriber) {
			this.channel = channel;
			this.subscriber = subscriber;
		}

		@Override
		public void run() {
		    while(true){
		        try {
		            LOGGER.info(String.format("subscribe redis, channel %s", channel.toString()));
		            JedisCluster cluster = JedisClusterFactory.getDefaultJedisCluster();
		            cluster.subscribe(subscriber, channel);
		            subscriber.unsubscribe(channel);
		            LOGGER.info(String.format("subscribe redis, channel %s 结束 ", channel.toString()));
		            
		        } catch (Exception e) {
		            LOGGER.error(String.format("receive redis channel %s message 发生异常", channel.toString()),e);
		        }
            }
		}

	}

	/**
	 * 实现redis订阅频道消息处理
	 * 
	 * @author liurunbin 2017年7月9日 cn.ivicar.redis.pubsub
	 */
	class RedisSubscriber extends JedisPubSub {
		public final ExecutorService threadPool = Executors.newFixedThreadPool(1);

		public RedisSubscriber() {
		}

		public void onMessage(final String channel, final String message) {
			LOGGER.info(String.format("receive redis published message, channel %s, message %s", channel, message));
			threadPool.execute(new Runnable() {
				
				@Override
				public void run() {
					// TODO Auto-generated method stub

					try {
						hander.handle(message);
						consumerStat.get(channel).increaseSuccess();
					} catch (Exception e) {
						e.printStackTrace();
						consumerStat.get(channel).increaseFail();
					} finally {
						consumerStat.get(channel).increaseTotal();
					}
				
				}
			});
			
		}

		public void onSubscribe(String channel, int subscribedChannels) {
			LOGGER.info(String.format("subscribe redis channel success, channel %s, subscribedChannels %d", channel,
					subscribedChannels));
		}

		public void onUnsubscribe(String channel, int subscribedChannels) {
			LOGGER.info(String.format("unsubscribe redis channel, channel %s, subscribedChannels %d", channel,
					subscribedChannels));

		}
	}

	class Stat {
		private AtomicLong success = new AtomicLong(0L);

		private AtomicLong fail = new AtomicLong(0L);

		private AtomicLong total = new AtomicLong(0L);

		public long getAndResetSuccess() {
			return this.success.getAndSet(0);
		}

		public long getAndResetFail() {
			return this.fail.getAndSet(0);
		}

		public long getTotal() {
			return this.total.get();
		}

		public void increaseSuccess() {
			this.success.incrementAndGet();
		}

		public void increaseFail() {
			this.fail.incrementAndGet();
		}

		public void increaseTotal() {
			this.total.incrementAndGet();
		}
	}

}
